package com.pactera.elasticsearch.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.pactera.elasticsearch.autoconfigure.ElasticsearchProperties;
import com.pactera.elasticsearch.bean.*;
import com.pactera.elasticsearch.commons.Elasticsearch;
import com.pactera.elasticsearch.commons.Page;
import com.pactera.elasticsearch.utils.Assert;
import com.pactera.elasticsearch.utils.SpringScanUtils;
import com.pactera.elasticsearch.utils.StrUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.Resource;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: liqingdian
 **/
public class ElasticServiceImpl implements ElasticService {

    private static Logger log = LoggerFactory.getLogger(ElasticServiceImpl.class);

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private ElasticsearchProperties properties;

    private List<Index> indexs;

    private void setIndexs() throws IOException {
        Set<Class<?>> classes = SpringScanUtils.doScan(properties.getPackageScan());
        indexs = classes.stream()
                .filter(a -> a.getAnnotation(com.pactera.elasticsearch.annocation.Index.class) != null)
                .map(a -> {
                    com.pactera.elasticsearch.annocation.Index annotation = a.getAnnotation(com.pactera.elasticsearch.annocation.Index.class);
                    String value = annotation.value();
                    String name = annotation.name();
                    String settings = annotation.settings();
                    String mappings = annotation.mappings();
                    Index index = new Index();
                    index.setValue(value);
                    index.setName(StringUtils.isBlank(name) ? value : name);
                    index.setSettings(settings);
                    index.setMappings(mappings);
                    return index;
                }).collect(Collectors.toList());
    }

    @Override
    public void initIndex() throws IOException {
        if (this.indexs == null) {
            this.setIndexs();
        }
        for (Index index : this.indexs) {
            String indexValue = index.getValue();
            GetIndexRequest get = new GetIndexRequest(indexValue);
            // 判断索引是否存在
            boolean exists = restHighLevelClient.indices().exists(get, RequestOptions.DEFAULT);
            if (exists) {
                log.info("{}({})已经存在，跳过初始化", index.getName(), index.getValue());
                continue;
            }
            CreateIndexRequest request = new CreateIndexRequest(indexValue);
            String settings = index.getSettings();
            if (StringUtils.isNotBlank(settings)) {
                Resource resource = SpringScanUtils.getResourceLoader().getResource("classpath:elasticsearch/" + settings);
                InputStream inputStream = resource.getInputStream();
                String text = StrUtils.inputStreamToString(inputStream);
                request.settings(text, XContentType.JSON);
            }
            String mappings = index.getMappings();
            if (StringUtils.isNotBlank(mappings)) {
                Resource resource = SpringScanUtils.getResourceLoader().getResource("classpath:elasticsearch/" + mappings);
                InputStream inputStream = resource.getInputStream();
                String text = StrUtils.inputStreamToString(inputStream);
                request.mapping(text, XContentType.JSON);
            }
            CreateIndexResponse response = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
            if (!response.isAcknowledged()) {
                log.warn("Elasticsearch初始化索引失败,请手动添加索引: {}", JSON.toJSONString(index));
            }
        }
    }

    @Override
    public List<Index> indices() throws IOException {
        if (this.indexs == null) {
            this.setIndexs();
        }
        return this.indexs;
    }

    @Override
    public void save(Elasticsearch document, boolean async) throws IOException {
        Assert.isNotNull(document, "[document]不能为空");
        String guid = document.getGuid();
        Assert.isNotBlank(guid, "[document.guid]不能为空");
        com.pactera.elasticsearch.annocation.Index annotation = document.getClass().getAnnotation(com.pactera.elasticsearch.annocation.Index.class);
        Assert.isNotNull(annotation, "document[@Index]注解缺失");
        String index = annotation.value();
        IndexRequest insert = new IndexRequest(index);
        insert.id(guid);
        document.setTimestamp(System.currentTimeMillis());
        insert.source(JSON.toJSONString(document,
                SerializerFeature.WriteMapNullValue,
                SerializerFeature.WriteDateUseDateFormat,
                SerializerFeature.WriteNullListAsEmpty,
                SerializerFeature.WriteNullStringAsEmpty,
                SerializerFeature.WriteBigDecimalAsPlain,
                SerializerFeature.WriteEnumUsingToString,
                SerializerFeature.SkipTransientField
                )
                , XContentType.JSON);
        if (async) {
            restHighLevelClient.indexAsync(insert, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
                @Override
                public void onResponse(IndexResponse indexResponse) {
                    log.info("{}-{}", indexResponse.getIndex(), indexResponse.getId());
                }

                @Override
                public void onFailure(Exception e) {
                    log.error(e.getMessage(), e);
                }
            });
        } else {
            restHighLevelClient.index(insert, RequestOptions.DEFAULT);
        }
    }

    @Override
    public void delete(String index, String guid) throws IOException {
        Assert.isNotBlank(index, "[index]不能为空");
        Assert.isNotBlank(guid, "[guid]不能为空");
        DeleteRequest delete = new DeleteRequest(index, guid);
        restHighLevelClient.delete(delete, RequestOptions.DEFAULT);
    }

    @Override
    public <T extends Elasticsearch> T info(String guid, Class<T> clazz) throws IOException {
        Assert.isNotBlank(guid, "[guid]不能为空");
        Assert.isNotNull(clazz, "[clazz]不能为空");
        com.pactera.elasticsearch.annocation.Index annotation = clazz.getAnnotation(com.pactera.elasticsearch.annocation.Index.class);
        Assert.isNotNull(annotation, "clazz的注解[@Index]缺失");
        String index = annotation.value();
        GetRequest find = new GetRequest(index, guid);
        GetResponse response = restHighLevelClient.get(find, RequestOptions.DEFAULT);
        if (response.isExists()) {
            byte[] sourceAsBytes = response.getSourceAsBytes();
            return JSON.parseObject(sourceAsBytes, clazz);
        }
        return null;
    }

    @Override
    public <T extends Elasticsearch> Page<T> pages(Search search, Class<T> clazz) throws IOException {
        Assert.isNotNull(search, "[search]不能为空");
        String index = search.getIndex();
        Assert.isNotBlank(index, "[index]不能为空");
        SearchSourceBuilder sourceBuilder = this.builderQuery(search);
        // 查询结果集
        SearchRequest searchRequest = new SearchRequest(index).source(sourceBuilder);
        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        List<T> list = new ArrayList<>();
        SearchHits hits = response.getHits();
        TotalHits totalHits = hits.getTotalHits();
        Page<T> pageInfo = new Page<>();
        pageInfo.setCurrent(search.getCurrent());
        pageInfo.setSize(search.getSize());
        // 查询总数
        long total = totalHits == null ? 0 : totalHits.value;
        pageInfo.setTotal(total);
        long size = search.getSize();
        long page = total / size;
        page = total % size > 0 ? page + 1 : page;
        pageInfo.setPages(page);
        for (SearchHit hit : hits.getHits()) {
            BytesReference sourceRef = hit.getSourceRef();
            String text = sourceRef.utf8ToString();
            T data = JSON.parseObject(text, clazz);
            list.add(data);
        }
        pageInfo.setRecords(list);
        return pageInfo;
    }

    /**
     * 构建查询条件
     *
     * @author: liqingdian
     */
    private SearchSourceBuilder builderQuery(Search search) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        int size = search.getSize().intValue();
        int from = (search.getCurrent().intValue() - 1) * size;
        // 构建分页
        sourceBuilder.from(from).size(size).timeout(new TimeValue(60, TimeUnit.SECONDS));
        // 构建排序
        List<Order> orders = search.getOrders();
        if (!CollectionUtils.isEmpty(orders)) {
            for (Order order : orders) {
                String name = order.getName();
                if (StringUtils.isNotBlank(name)) {
                    boolean asc = order.getAsc() == null ? false : order.getAsc();
                    FieldSortBuilder builder = new FieldSortBuilder(order.getName()).order(asc ? SortOrder.ASC : SortOrder.DESC);
                    sourceBuilder.sort(builder);
                }
            }
        } else {
            sourceBuilder.sort(new FieldSortBuilder("timestamp").order(SortOrder.DESC));
        }
        // 构建查询字段
        String[] includes = search.getIncludes();
        if (ArrayUtils.isNotEmpty(includes)) {
            sourceBuilder.fetchSource(includes, new String[0]);
        }
        // 构建查询条件
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        Query query = search.getQuery();
        if (query != null) {
            // fuzzy模糊查询
            List<Fuzzy> fuzzies = query.getFuzzies();
            if (!CollectionUtils.isEmpty(fuzzies)) {
                for (Fuzzy fuzzy : fuzzies) {
                    FuzzyQueryBuilder fuzzyQueryBuilder = QueryBuilders.fuzzyQuery(fuzzy.getName(), fuzzy.getValue());
                    this.matched(queryBuilder, fuzzyQueryBuilder, fuzzy.getMatched());
                }
            }
            // match全文查询-分词（邻近）
            List<Match> matches = query.getMatches();
            if (!CollectionUtils.isEmpty(matches)) {
                for (Match match : matches) {
                    MatchPhraseQueryBuilder matchPhraseQueryBuilder = QueryBuilders.matchPhraseQuery(match.getName(), match.getValue());
                    this.matched(queryBuilder, matchPhraseQueryBuilder, match.getMatched());
                }
            }
            // range区间过滤
            List<Range> ranges = query.getRanges();
            if (!CollectionUtils.isEmpty(ranges)) {
                for (Range range : ranges) {
                    RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(range.getName()).gte(range.getGte()).lte(range.getLte());
                    if (StringUtils.isNotBlank(range.getFormat())) {
                        rangeQueryBuilder.format(range.getFormat());
                    }
                    this.matched(queryBuilder, rangeQueryBuilder, range.getMatched());
                }
            }
            // terms多值精确查询-不分词
            List<Term> terms = query.getTerms();
            if (!CollectionUtils.isEmpty(terms)) {
                for (Term term : terms) {
                    TermsQueryBuilder termsQueryBuilder = QueryBuilders.termsQuery(term.getName(), term.getValues());
                    this.matched(queryBuilder, termsQueryBuilder, term.getMatched());
                }
            }
        }
        return sourceBuilder.query(queryBuilder);
    }

    /**
     * 查询匹配：1：must,2:filter,3:should,4:must_not
     *
     * @author: liqingdian
     */
    private void matched(BoolQueryBuilder query, QueryBuilder builder, int matched) {
        switch (matched) {
            case 1:
                query.must(builder);
                break;
            case 2:
                query.filter(builder);
                break;
            case 3:
                query.should(builder);
                break;
            case 4:
                query.mustNot(builder);
                break;
            default:
                query.filter(builder);
        }
    }
}
